package com.beta.rabbit.producer.broker;

import com.beta.rabbit.api.Message;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;

/**
 * 具体发送不同种类型的消息
 */
@Slf4j
public class RabbitBrokerImpl implements RabbitBroker {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Override
    public void rapidSend(Message message) {
        sendKernel(message);
    }

    @Override
    public void confirmSend(Message message) {

    }

    @Override
    public void reliantSend(Message message) {

    }

    @Override
    public void sendMessages() {

    }

    /**
     * 发送消息的核心方法，使用异步线程池发送消息
     * @param message
     */
    void sendKernel(Message message){

        AsyncQueue.submit((Runnable)() -> {

            // exchange    //? rabbitTemplate.getExchange()
            String topic = message.getTopic();
            String routingKey = message.getRoutingKey();
            CorrelationData correlationData =
                    new CorrelationData(String.format("%s#%s",
                            message.getMessageId(),
                            System.currentTimeMillis()));

            rabbitTemplate.convertAndSend(topic, routingKey, message, correlationData);
            log.info("#RabbitBrokerImpl.sendKernel# send to rabbitmq, messageId: {}", message.getMessageId());

        });
    }
}
